package com.atguigu.userprofile.service.impl;

import com.alibaba.fastjson.JSON;
import com.atguigu.userprofile.bean.TagCondition;
import com.atguigu.userprofile.bean.TagInfo;
import com.atguigu.userprofile.bean.UserGroup;
import com.atguigu.userprofile.constants.ConstCodes;
import com.atguigu.userprofile.mapper.UserGroupMapper;
import com.atguigu.userprofile.service.TagInfoService;
import com.atguigu.userprofile.service.UserGroupService;
import com.atguigu.userprofile.utils.RedisUtil;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;

import org.apache.catalina.User;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import redis.clients.jedis.Jedis;

import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author zhangchen
 * @since 2021-05-04
 */
@Service
@Slf4j
@DS("mysql")
public class UserGroupServiceImpl extends ServiceImpl<UserGroupMapper, UserGroup> implements UserGroupService {


    /*
            id： 写入mysql，会自动生成
            condition_json_str: 把List<TagCondition> tagConditions条件转换为json格式。
            condition_comment: 用一句话描述传入的条件。调用UserGroup.conditionJsonToComment()生成
            update_time:  更新人群时，生成。 目前无需生成。
            create_time:  第一次写入mysql时，生成。需要生成

            额外计算userGroupNum: 查询clickhouse
     */
    @Override
    public void saveUserGroup(UserGroup userGroup) {

        String condition_json_str = JSON.toJSONString(userGroup.getTagConditions());
        userGroup.setConditionJsonStr(condition_json_str);

        userGroup.setConditionComment(userGroup.conditionJsonToComment());
        userGroup.setCreateTime(new Date());

        //写入mysql
        save(userGroup);
    }


    @Override
    public void writeUserGroupToCk(UserGroup userGroup) {

        //准备sql
        String bitmapAndSql = generateBitmapAndSql(userGroup);
        //调动Dao
        getBaseMapper().writeUserGroupToCk(userGroup.getId().toString(),bitmapAndSql);
    }

    /*
        将分群包写入redis

        key:  字符串，唯一。 用分群id作为key
        value:
                    单值: string,hash
                    多值: list,set,zset,hash
                        set集合提供api: sismenber，可以判断一个元素是否在集合，符合需求！

         数据格式:  用户分群1=[1,2,3,4,5]
     */
    @Override
    public int writeUserGroupToRedis(UserGroup userGroup) {

        //查询[1,2,3,4,5]
        List<String> uids = getBaseMapper().getUserGroupById(userGroup.getId().toString());

        //把List转换为String[]
        String [] value = new String[]{};
        value = uids.toArray(value);

        //准备key
        String key = "userGroup_"+ userGroup.getId();

        Jedis jedis = RedisUtil.getJedis();

        //先删除数据，再写入，保证更新的准确性
        jedis.del(key);
        //使用连接    更新前：  usergroup_1: [1,2,3]   更新后从ck中查询:   usergroup_1: [2,3,4]
        jedis.sadd(key,value);

        jedis.close();

        return uids.size();
    }

    @Override
    public int evaluateUserGroup(UserGroup userGroup) {

        String sql = generateBitmapAndSql(userGroup);

        int num = getBaseMapper().evaluateUserGroup(sql);

        return num;
    }

    /*
        1.先删除旧的信息
        2.再写入新的信息
     */
    @Override
    public void updateUserGroupToCk(UserGroup userGroup) {
        //删除
        getBaseMapper().deleteUserGroupById(userGroup.getId().toString());
        //写入新的
        writeUserGroupToCk(userGroup);
    }

    @Autowired
    private TagInfoService tagInfoService;
    /*
            完成最终的用户分群查询sql
                两两拼接 bitmapAnd()

            bitmapAnd(
                  bitmapAnd( (sql1), (sql2) ),
                    ( sql3 )
                )
     */
    public String generateBitmapAndSql(UserGroup userGroup){

        //获取所有的查询条件
        List<TagCondition> tagConditions = userGroup.getTagConditions();

        //获取业务日期
        String doDate = userGroup.getBusiDate();

        //查询所有三级标签的元数据
        Map<String, TagInfo> tags = tagInfoService.getTagInfoMapWithCode();

        StringBuilder result = new StringBuilder();

        for (TagCondition tagCondition : tagConditions) {

            //第一个条件，必须在外层嵌套()  sql3
            String sql = generateSingleTagConditionSql(tagCondition, tags, doDate);

            //   (sql1)
            if (result.length() == 0){
                result.append("(").append(sql).append(")");
            }else{
                //  bitmapAnd(  bitmapAnd( (sql1), (sql2) ) , (sql3))
                result.insert(0,"bitmapAnd(")
                      .append(", (")
                      .append(sql)
                      .append("))");
            }
        }

        return result.toString();

    }

    /*
     根据传入的单个 TagCondition，生成单条用户分群的查询:

    select
        groupBitmapMergeState(us)
     from ①user_tag_value_string
      where dt = ②'2020-06-14'
            and tag_code = ③'tag_population_attribute_nature_period'
           and tag_value ④in  ⑤('90后', '00后')


           ①： 前台传入的只有标签的英文名，没有标签值的类型。需要从后台查询每个标签对应的 TagInfo，再获取标签值的类型。
                    提前把所有标签的英文名及对应的TagInfo，缓存在一个Map集合中
                        Map<String,TagInfo>
                            调用:  TagInfoServiceImpl.getTagInfoMapWithCode()

           ②:业务日期
           ③： 标签名
           ④操作符:  需要把传入的操作符转换为sql中的运算符
                        传入 的是 eq, 转换为 sql中 的 =

           ⑤标签值:
                        如果值是字符串或日期，需要为每个值添加 '
                        如果是多个值，两两拼接 ,
                        如果操作符是范围查询(in,notin),还需要在最外层拼接()

     */
    public String generateSingleTagConditionSql(TagCondition condition,Map<String,TagInfo> tags,String doDate){

        //查询的标签
        TagInfo tagInfo = tags.get(condition.getTagCode());

        String sourceTable = "";
        switch (tagInfo.getTagValueType()){
            case ConstCodes.TAG_VALUE_TYPE_LONG : sourceTable = "user_tag_value_long"; break;
            case ConstCodes.TAG_VALUE_TYPE_DECIMAL : sourceTable = "user_tag_value_decimal"; break;
            case ConstCodes.TAG_VALUE_TYPE_DATE : sourceTable = "user_tag_value_date"; break;
            case ConstCodes.TAG_VALUE_TYPE_STRING : sourceTable = "user_tag_value_string"; break;
        }

        String sqlOperator = getConditionOperator(condition.getOperator());

        String values = "";
        //拼接查询的值
        if (ConstCodes.TAG_VALUE_TYPE_STRING.equals(tagInfo.getTagValueType()) ||
            ConstCodes.TAG_VALUE_TYPE_DATE.equals(tagInfo.getTagValueType())
          ){
            // ' a ',' b ',' c '
            values = "'"+StringUtils.join( condition.getTagValues(),"','")+"'";

        }else {
            values = StringUtils.join( condition.getTagValues(),",");

        }

        //如果操作符是范围查询(in,notin),还需要在最外层拼接()
        if ( "in".equals(condition.getOperator()) || "nin".equals(condition.getOperator())){
            values = "( " + values +" )";
        }

        //声明模版
        String template = "  select groupBitmapMergeState(us) from %s where dt = '%s'  and tag_code =  '%s' " +
                          "   and tag_value %s  %s  ";

        String sql = String.format(template, sourceTable, doDate, condition.getTagCode().toLowerCase(), sqlOperator, values);
        return sql;

    }

    //把前台传入的字符串操作符转为 sql中的运算符
    private String getConditionOperator(String operator){

        switch (operator) {
            case "eq" : return "=";
            case "neq" : return "<>";
            case "gte" : return ">=";
            case "lte" : return "<=";
            case "gt" : return ">";
            case "lt" : return "<";
            case "in" : return "in";
            case "nin" : return "not in";
        }
        throw new RuntimeException("没有匹配的操作符!");

    }
}
